package org.groupg.filetodb;

import com.univocity.parsers.common.ParsingContext;
import com.univocity.parsers.common.processor.RowProcessor;
import com.univocity.parsers.csv.CsvParser;
import com.univocity.parsers.csv.CsvParserSettings;

import java.io.File;
import java.util.concurrent.atomic.AtomicLong;

public class LargeCsvStreamReader {

    // 内存监控线程
    private static class MemoryMonitor implements Runnable {
        private final AtomicLong running;
        private final long maxMemoryBytes;
        private volatile boolean stop = false;

        public MemoryMonitor(AtomicLong running, long maxMemoryBytes) {
            this.running = running;
            this.maxMemoryBytes = maxMemoryBytes;
        }

        public void stop() {
            this.stop = true;
        }

        @Override
        public void run() {
            Runtime runtime = Runtime.getRuntime();
            while (!stop) {
                long usedMemory = runtime.totalMemory() - runtime.freeMemory();
                if (usedMemory > maxMemoryBytes) {
                    System.err.println("内存使用超过限制: " + (usedMemory / 1024 / 1024) + "MB");
                    System.exit(1);
                }
                running.set(usedMemory);
                try {
                    Thread.sleep(100); // 每100ms检查一次
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    // 行处理器
    private static class MyRowProcessor implements RowProcessor {
        private final AtomicLong memoryUsage;
        private long rowCount = 0;

        public MyRowProcessor(AtomicLong memoryUsage) {
            this.memoryUsage = memoryUsage;
        }

        @Override
        public void processStarted(ParsingContext context) {
            System.out.println("开始解析文件...");
        }

        @Override
        public void rowProcessed(String[] row, ParsingContext context) {
            // 处理每一行数据 - 这里只是示例
            if (rowCount % 100000 == 0) {
                System.out.printf("已处理 %,d 行, 当前内存使用: %,d MB%n",
                        rowCount, memoryUsage.get() / 1024 / 1024);
            }
            rowCount++;
            System.out.println(row[0]);
            // 实际业务处理逻辑放在这里
            // 注意: 不要在此方法中累积数据，否则会内存溢出
        }

        @Override
        public void processEnded(ParsingContext context) {
            System.out.printf("解析完成! 共处理 %,d 行%n", rowCount);
        }
    }

    public static void main(String[] args) {
        // 配置参数
        String filePath = "E:\\Develop\\demo001-4kw.csv";
        long maxMemoryBytes = 1024 * 1024 * 1024; // 1GB

        // 内存监控
        AtomicLong memoryUsage = new AtomicLong();
        MemoryMonitor monitor = new MemoryMonitor(memoryUsage, maxMemoryBytes);
        Thread monitorThread = new Thread(monitor);
        monitorThread.setDaemon(true);
        monitorThread.start();

        try {
            // 配置解析器设置
            CsvParserSettings settings = new CsvParserSettings();

            // 配置行处理器 (流式处理)
            settings.setProcessor(new MyRowProcessor(memoryUsage));

            // 性能优化设置
            settings.setLineSeparatorDetectionEnabled(true); // 自动检测行分隔符
            settings.setMaxCharsPerColumn(100000); // 设置每列最大字符数
            settings.setNullValue(""); // 空值处理
            settings.setEmptyValue(""); // 空字符串处理
            settings.setAutoClosingEnabled(true); // 自动关闭输入流

            // 内存优化设置
            settings.setReadInputOnSeparateThread(false); // 单线程处理更省内存
            settings.setInputBufferSize(65536); // 64KB缓冲区

            // 创建解析器
            CsvParser parser = new CsvParser(settings);

            // 开始解析
            parser.parse(new File(filePath));

        } catch (Exception e) {
            System.err.println("解析失败: " + e.getMessage());
            e.printStackTrace();
        } finally {
            monitor.stop();
        }
    }
}
